package com.atguigu.wordcount.reducerjoin;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.metrics.util.MBeanUtil;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @ClassName ReducerJoinReducer
 * @DeScription ReducerJoinReducer
 * @Author 剧情再美终是戏
 * @Date 2019/11/20 11:43
 * @Version 1.0
 **/
public class ReducerJoinReducer extends Reducer<NullWritable, JoinBean, NullWritable, JoinBean> {

    private List<JoinBean> orderList = new ArrayList();

    private Map<String, String> paMap = new HashMap();

    @Override
    protected void reduce(NullWritable key, Iterable<JoinBean> values, Context context) throws IOException, InterruptedException {
        for (JoinBean value : values) {
            String source = value.getSource();
            if ("order.txt".equals(source)) {
                JoinBean newJoinBean = new JoinBean();
                try {
                    BeanUtils.copyProperties(newJoinBean, value);
                    orderList.add(newJoinBean);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
            } else {
                paMap.put(value.getpId(), value.getpName());
            }
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        for (JoinBean joinBean : orderList) {
            joinBean.setpName(paMap.get(joinBean.getpId()));
            context.write(NullWritable.get(), joinBean);
        }
    }
}
